package com.surfilter.mq.rocketmq;

import com.alibaba.rocketmq.client.consumer.DefaultMQPullConsumer;
import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.PullResult;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.client.exception.MQBrokerException;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.alibaba.rocketmq.common.message.MessageQueue;
import com.alibaba.rocketmq.common.protocol.heartbeat.MessageModel;
import com.alibaba.rocketmq.remoting.exception.RemotingException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.*;

public class ConsumerManage {
	
	private static Logger logger = LoggerFactory.getLogger(ConsumerManage.class);

	/**
	 * 该方法只处理一个Topic
	 * @param config
	 * @param service 数据处理服务
	 * @throws MQClientException
	 */
	public static void openConsumer(RocketMqConfig config, final OprateMessageService service) throws MQClientException{
		DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(config.getGroup());
		consumer.setMessageModel(MessageModel.CLUSTERING);
		consumer.setNamesrvAddr(config.getNamesrvAddr());
		consumer.setInstanceName(config.getInstanceName());
		consumer.subscribe(config.getTopic(), config.getTags());
		consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
		consumer.setConsumeMessageBatchMaxSize(config.getBatchSize());
		consumer.registerMessageListener(new MessageListenerConcurrently() {
			public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
				List<byte[]> messages = new ArrayList<byte[]>();
				for(MessageExt msg : msgs){
					messages.add(msg.getBody());
				}
				service.oprate(messages);
				return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
			}
		});
		consumer.start();
	}
	
	/**
	 * 
	 * @param config
	 * @param service
	 * @return
	 */
	public static boolean openConsumerA(RocketMqConfig config, final OprateMessageService service){
		boolean result = false;
		try{
			openConsumer(config, service);
			result = true;
		}catch(Exception e){
			logger.error("打开消费者:{}失败", config.getInstanceName(), e);
		}
		return result;
	}
	
	/**
	 * 
	 * @param config
	 * @param listener
	 * @return
	 */
	public static boolean openConsumerB(RocketMqConfig config, final MessageListenerConcurrently listener){
		boolean result = false;
		try{
			openConsumer(config, listener);
			result = true;
		}catch(Exception e){
			logger.error("打开消费者:{}失败", config.getInstanceName(), e);
		}
		return result;
	}
	
	
	/**
	 * 该方法只处理一个Topic
	 * @param config
	 * @param listener 消息监听
	 * @throws MQClientException
	 */
	public static void openConsumer(RocketMqConfig config, MessageListenerConcurrently listener) throws MQClientException{
		DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(config.getGroup());
		consumer.setNamesrvAddr(config.getNamesrvAddr());
		consumer.setInstanceName(config.getInstanceName());
		consumer.subscribe(config.getTopic(), config.getTags());
		consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
		consumer.setConsumeMessageBatchMaxSize(config.getBatchSize());
		consumer.registerMessageListener(listener);
		consumer.start();
	}
	
	public static void openConsumerPull(RocketMqConfig config, OprateMessageService service){
		try{
			DefaultMQPullConsumer consumer = new DefaultMQPullConsumer(config.getGroup());
			consumer.setMessageModel(MessageModel.CLUSTERING);
			consumer.setNamesrvAddr(config.getNamesrvAddr());
			consumer.start();
			new ConsumerPullMesgQueueThread(config.getTopic(), consumer, service, config.getBatchSize()).start();
		}catch(Exception e){
			logger.error("启动Pull消费者异常", e);
		}
	}
	
//	public static void main(String[] args) throws MQClientException {
//		String group = PropertiesUtils.getStringValue("distinct.group", "group");
//		String topic = "test_topic";
//		RocketMqConfig config = new RocketMqConfig(group, topic, "1", "192.168.60.152:9876", "intance_name", 1);
//		ConsumerManage.openConsumerPull(config, new OprateMessageServiceImpl());
//	}
	
	public enum ConsumerType{
		PULL,PUSH
	}
	
}

class ConsumerPullMesgQueueThread extends Thread{
	
	private Logger logger = LoggerFactory.getLogger(ConsumerPullMesgQueueThread.class);
	
	private String topic;
	
	private Map<MessageQueue, Long> queue_offsets = new HashMap<MessageQueue, Long>();
	
	private DefaultMQPullConsumer consumer;
	
	private OprateMessageService service;
	
	private int maxNums = 1;
	
	public ConsumerPullMesgQueueThread(String topic, DefaultMQPullConsumer consumer,
			OprateMessageService service, int maxNums) {
		this.topic = topic;
		this.consumer = consumer;
		this.service = service;
		this.maxNums = maxNums;
	}

	@Override
	public void run() {
		// TODO Auto-generated method stub
		while(true){
			try{
				Thread.sleep(3000);
				initMessageQueue();
				consumeMesg();
			}catch(Exception e){
				logger.error("Pull Consumer 消费：{}数据异常", topic, e);
			}
		}
	}
	
	public void initMessageQueue() throws MQClientException{
		try{
			Set<MessageQueue> queues = consumer.fetchSubscribeMessageQueues(topic);
			if(queues == null || queues.isEmpty()){
				queue_offsets.clear();
				return;
			}
			for(MessageQueue queue : queues){
				if(!queue_offsets.containsKey(queue))queue_offsets.put(queue, 0l);
			}
			Set<MessageQueue> queue_keys = new HashSet<MessageQueue>();
			queue_keys.addAll(queue_offsets.keySet());
			for(MessageQueue queue : queue_keys){
				if(!queues.contains(queue)){
					queue_offsets.remove(queue);
				}
			}
		}catch(Exception e){
			throw new MQClientException("初始化所有消息队列异常", e);
		}
		
	}
	
	public void consumeMesg(){
		if(queue_offsets == null || queue_offsets.isEmpty())return;
		for(MessageQueue queue : queue_offsets.keySet()){
			try{
				oprateQueueMesg(queue);
			}catch(Exception e){
				logger.error("Topic:{} QueueId:{} BrokerName:{} 拉取数据异常", topic, queue.getQueueId(), queue.getBrokerName(), e);
			}
		}
	}
	
	public void oprateQueueMesg(MessageQueue queue) throws MQClientException, RemotingException, MQBrokerException, InterruptedException{
		SINGLE_MQ : while(true){
			PullResult result = consumer.pull(queue, null, queue_offsets.get(queue), maxNums);
			long next_offset = result.getNextBeginOffset();
			List<MessageExt> mesg_ext = result.getMsgFoundList();
			if(mesg_ext == null || mesg_ext.isEmpty()) return;
			List<byte[]> messages = new ArrayList<byte[]>();
			for(MessageExt msg : mesg_ext){
				messages.add(msg.getBody());
			}
			service.oprate(messages);
			queue_offsets.put(queue, next_offset);
			switch (result.getPullStatus()) {  
            case FOUND://pullSataus  
                break;  
            case NO_MATCHED_MSG:  
                break;  
            case NO_NEW_MSG:  
                break SINGLE_MQ;
            case OFFSET_ILLEGAL:
                break;
            default:
                break; 
			}
		}
	}

	public String getTopic() {
		return topic;
	}

	public void setTopic(String topic) {
		this.topic = topic;
	}

	public Map<MessageQueue, Long> getQueue_offsets() {
		return queue_offsets;
	}

	public void setQueue_offsets(Map<MessageQueue, Long> queue_offsets) {
		this.queue_offsets = queue_offsets;
	}

	public DefaultMQPullConsumer getConsumer() {
		return consumer;
	}

	public void setConsumer(DefaultMQPullConsumer consumer) {
		this.consumer = consumer;
	}
	
}
